package com.galeno.day07;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/2320:41
 */
public class IncomeCount {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",22222);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        env.enableCheckpointing(10000);
        DataStreamSource<String> lines = env.socketTextStream("galeno", 9999);
        SingleOutputStreamOperator<Tuple3<String, String, Double>> maped = lines.map(new MapFunction<String, Tuple3<String, String, Double>>() {
            @Override
            public Tuple3<String, String, Double> map(String value) throws Exception {
                String[] split = value.split(",");
                return Tuple3.of(split[0], split[1], Double.parseDouble(split[2]));
            }
        });
        KeyedStream<Tuple3<String, String, Double>, String> keyedStream = maped.keyBy(x -> x.f0);
        SingleOutputStreamOperator<Tuple4<String, Double, String, Double>> streamOperator = keyedStream.process(new KeyedProcessFunction<String, Tuple3<String, String, Double>, Tuple4<String, Double, String, Double>>() {
            private MapState<String, Double> mapState;

            @Override
            public void open(Configuration parameters) throws Exception {
                MapStateDescriptor<String, Double> mapStateDescriptor = new MapStateDescriptor<>("map-state", String.class, Double.class);
                mapState = getRuntimeContext().getMapState(mapStateDescriptor);


            }

            @Override
            public void processElement(Tuple3<String, String, Double> value, Context ctx, Collector<Tuple4<String, Double, String, Double>> out) throws Exception {
                String city = value.f1;
                Double money = value.f2;
                Double historyIncome = mapState.get(city);
                if (historyIncome == null) {
                    historyIncome = 0.0;
                }
                historyIncome += money;

                mapState.put(city, historyIncome);

                Iterable<Double> doubleIterable = mapState.values();
                Double totalIncome = 0.0;
                for (Double aDouble : doubleIterable) {
                    totalIncome += aDouble;
                }

                out.collect(Tuple4.of(value.f0, totalIncome, city, historyIncome));


            }
        });
        streamOperator.print();
        env.execute();


    }
}
